package com.lili.collection.mq.group;

import com.lili.collection.core.config.ProcessConfig;
import com.lili.collection.core.context.BatchRecordContext;
import com.lili.collection.core.context.Context;
import com.lili.collection.core.context.PipeContext;
import com.lili.collection.core.context.StageContext;
import com.lili.collection.core.context.impl.GenericBatchRecordContext;
import com.lili.collection.core.domain.RunnerMsgTransfer;
import com.lili.collection.core.error.PluginRuntimeException;
import com.lili.collection.core.error.StageRecordException;
import com.lili.collection.core.event.StageEvent;
import com.lili.collection.core.event.StageEventListener;
import com.lili.collection.core.event.StageEventListenerRegister;
import com.lili.collection.core.handler.IExecValidateHandler;
import com.lili.collection.core.plugins.ProcessPlugin;
import com.lili.collection.core.runner.StageGroupRunner;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.List;

/**
 * 通用输入Runner
 *
 * @author lili
 * @date 2023/3/7 11:18
 */
@Slf4j
public class AsyncProcessorGroupRunner implements StageGroupRunner, StageEventListenerRegister {

    protected List<StageEventListener> stageContextChangedEventListener;

    @Getter
    private PipeContext pipeContext;
    private List<ProcessPlugin> processPluginList;
    @Getter
    private RunnerMsgTransfer transfer;
    @Override
    public void setRunnerTransfer(RunnerMsgTransfer transfer) {
        this.transfer=transfer;
    }

    public AsyncProcessorGroupRunner(PipeContext pipeContext) {
        this.pipeContext = pipeContext;
    }

    @Override
    public void pause() {

    }

    @Override
    public void resume() {

    }

    @Override
    public void stop() {

    }

    @Override
    public void dispose() {
    }

    private void ensurePlugin() {
        if (this.processPluginList != null) return;
        this.processPluginList=new ArrayList<>();
        List<ProcessConfig> processConfigList = this.pipeContext.getContext().getProcessConfigList();
        for (ProcessConfig processConfig : processConfigList) {
            ProcessPlugin plugin = processConfig.getPlugin();
            plugin.init(processConfig);
            plugin.setPipeContext(this.pipeContext);
            this.processPluginList.add(plugin);
        }
    }

    @Override
    public void run() {
        try {
            this.transfer.addThread(Thread.currentThread());
            this.ensurePlugin();
            long execNum=0;
            while(true){
                if(this.transfer.isRunError()){
                    break;
                }
                execNum = this.execValidate(execNum);
                StageContext stageContext = this.getPipeContext().getExchangeStageContext().take(StageType.Input);
                //处理数据
                for (ProcessPlugin processPlugin : processPluginList) {
                    processPlugin.execute(stageContext);
                }
                this.getPipeContext().getExchangeStageContext().put(StageType.ProcessGroup, stageContext);
                triggerStageEvent(StageEvent.STAGE_CONTEXT_CHANGE, this.getStageType(), stageContext);
                if(stageContext.getStageContextState() == StageContext.StageContextState.COMPLETE){
                    break;
                }
            }
        } catch (Exception e) {
            this.transfer.setRunError(true);
            this.transfer.interruptRunner();
            BatchRecordContext recordContext = new GenericBatchRecordContext();
            recordContext.setStageRecordState(StageContext.StageRecordState.ALL_ERROR);
            recordContext.setException(new StageRecordException(e));
            this.getPipeContext().getExchangeStageContext().put(StageType.ProcessGroup, recordContext);
            log.error("运行异步处理组错误!",e);
        }finally {
            if(Thread.currentThread().isInterrupted()){
                Thread.interrupted();
            }
            this.transfer.setProcessExit(true);
        }
    }

    private long execValidate(long execNum) {
        Context context = this.pipeContext.getContext();
        if(context.getStopSignal()){
            throw new PluginRuntimeException("采集强制停止异常！");
        }
        if((execNum++)%100==0){
            IExecValidateHandler validateHandler = context.getValidateHandler();
            if(validateHandler!=null){
                if(!validateHandler.onValidate()){
                    throw new PluginRuntimeException("任务执行校验失败！");
                }
            }
        }
        return execNum;
    }

    @Override
    public StageType getStageType() {
        return StageType.ProcessGroup;
    }

    @Override
    public synchronized void registerStageEventListener(StageEventListener listener) {
        ensureStageEventListener();
        this.stageContextChangedEventListener.add(listener);
    }

    @Override
    public void triggerStageEvent(StageEvent stageEvent, StageType stageType, Object params) {
        this.stageContextChangedEventListener.forEach(r -> r.onStageEventTrigger(stageEvent, this.getStageType(), params));
    }

    private void ensureStageEventListener() {
        if (this.stageContextChangedEventListener == null)
            this.stageContextChangedEventListener = new ArrayList<>();
    }
}
